/**
 * @title: KuduPartitionTest
 * @projectName kuduMaster
 * @description: TODO
 * @author Administrator
 * @date 2019/11/1119:17
 */

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.CreateTableOptions;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.PartialRow;
import org.junit.Before;
import org.junit.Test;

import java.util.LinkedList;
import java.util.List;

/**
 * @author Administrator
 * @title: KuduPartitionTest
 * @description: TODO
 * @date 2019/11/11  19:17
 */
public class KuduPartitionTest {


    //声明全局变量 KuduClient后期通过它来操作kudu表
    private KuduClient kuduClient;
    //指定kuduMaster地址
    private String kuduMaster;
    //指定表名
    private String tableName;


    @Before
    public void init() {
        //初始化操作
        kuduMaster = "node01:7051,node02:7051,node03:7051";
        //指定表名
        tableName = "student";
        KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(kuduMaster);
        kuduClientBuilder.defaultSocketReadTimeoutMs(10000);
        kuduClient = kuduClientBuilder.build();

    }

    private static ColumnSchema newColumn(String name, Type type, boolean iskey) {
        ColumnSchema.ColumnSchemaBuilder column = new
                ColumnSchema.ColumnSchemaBuilder(name, type);
        column.key(iskey);
        return column.build();
    }

    /**
     * 范围分区：
     * RangePartition
     */
    @Test
    public void testRangePartition() throws KuduException {
        LinkedList<ColumnSchema> columnSchemas = new LinkedList<ColumnSchema>();
        columnSchemas.add(newColumn("CompanyId", Type.INT32, true));
        columnSchemas.add(newColumn("WorkId", Type.INT32, false));
        columnSchemas.add(newColumn("Name", Type.STRING, false));
        columnSchemas.add(newColumn("Gender", Type.STRING, false));
        columnSchemas.add(newColumn("Photo", Type.STRING, false));

        Schema schema = new Schema(columnSchemas);
        CreateTableOptions tableOptions = new CreateTableOptions();
        //设置副本数
        tableOptions.setNumReplicas(1);
        //设置范围分区的规则
        LinkedList<String> parcols = new LinkedList<String>();
        parcols.add("CompanyId");
        //设置按照那个字段进行range分区
        tableOptions.setRangePartitionColumns(parcols);
        int count = 0;
        for (int i = 0; i < 10; i++) {
            PartialRow lower = schema.newPartialRow();
            lower.addInt("CompanyId", count);

            PartialRow upper = schema.newPartialRow();
            count += 10;
            upper.addInt("CompanyId", count);
            tableOptions.addRangePartition(lower, upper);
        }
        kuduClient.createTable("student", schema, tableOptions);
        kuduClient.close();
    }

    /**
     * hash分区
     */
    @Test
    public void testHashPartition() throws KuduException {
//设置表的schema
        LinkedList<ColumnSchema> columnSchemas = new LinkedList<ColumnSchema>();
        columnSchemas.add(newColumn("CompanyId", Type.INT32, true));
        columnSchemas.add(newColumn("WorkId", Type.INT32, false));
        columnSchemas.add(newColumn("Name", Type.STRING, false));
        columnSchemas.add(newColumn("Gender", Type.STRING, false));
        columnSchemas.add(newColumn("Photo", Type.STRING, false));

        Schema schema = new Schema(columnSchemas);
        CreateTableOptions tableOptions = new CreateTableOptions();
        tableOptions.setNumReplicas(1);
        //设置范围分区的规则
        LinkedList<String> parcols = new LinkedList<String>();
        parcols.add("CompanyId");
        //设置按照那个字段进行range分区
        tableOptions.addHashPartitions(parcols, 6);

        kuduClient.createTable("dog", schema, tableOptions);
        kuduClient.close();

    }

    /**
     * 测试分区：
     * 多级分区
     * Multilevel Partition
     * 混合使用hash分区和range分区
     * <p>
     * 哈希分区有利于提高写入数据的吞吐量，而范围分区可以避免tablet无限增长问题，
     * hash分区和range分区结合，可以极大的提升kudu的性能
     */
    @Test
    public void testMultilevelPartition() throws KuduException {
        //设置表的schema
        LinkedList<ColumnSchema> columnSchemas = new LinkedList<ColumnSchema>();
        columnSchemas.add(newColumn("CompanyId", Type.INT32, true));
        columnSchemas.add(newColumn("WorkId", Type.INT32, false));
        columnSchemas.add(newColumn("Name", Type.STRING, false));
        columnSchemas.add(newColumn("Gender", Type.STRING, false));
        columnSchemas.add(newColumn("Photo", Type.STRING, false));

        Schema schema = new Schema(columnSchemas);
        CreateTableOptions tableOptions = new CreateTableOptions();
        tableOptions.setNumReplicas(1);
        List<String> parcols = new LinkedList<String>();
        parcols.add("CompanyId");
        tableOptions.addHashPartitions(parcols, 5);
        int count = 0;
        for (int i = 0; i < 10; i++) {
            PartialRow lower = schema.newPartialRow();
            lower.addInt("CompanyId", count);
            count += 10;
            PartialRow upper = schema.newPartialRow();
            upper.addInt("CompanyId", count);
            tableOptions.addRangePartition(lower, upper);
        }
        kuduClient.createTable("cat", schema, tableOptions);
        kuduClient.close();
    }
}
